package com.cdream.basic.thread;

import cn.hutool.core.thread.ThreadUtil;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

@FunctionalInterface
interface RejectStrategy<T> {
    void reject(BlockQueue<T> queue, T t);
}

/**
 * @author cdream
 * @date 2022/2/15
 */
@Slf4j
public class TestPool {
    public static void main(String[] args) {
        ThreadPool threadPool = new ThreadPool(2, 5, TimeUnit.SECONDS, 2, (queue, r) -> {
            ThreadUtil.sleep(3000);
            throw new RuntimeException("超时了");
        });
        for (int i = 0; i < 5; i++) {
            threadPool.execute(() -> {
                ThreadUtil.sleep(2000);
                log.debug("hahaha");
            });
        }

    }

}

class ThreadPool {
    private final HashSet<Worker> workers = new HashSet<>();
    private BlockQueue<Runnable> taskQueue;
    private int coreSize;

    private RejectStrategy<Runnable> rejectStrategy;

    private int timeout;

    private TimeUnit timeUnit;

    public ThreadPool(int coreSize, int timeout, TimeUnit timeUnit, int capacity, RejectStrategy<Runnable> strategy) {
        this.coreSize = coreSize;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
        this.taskQueue = new BlockQueue<>(capacity);
        this.rejectStrategy = strategy;
    }

    public void execute(Runnable runnable) {
        synchronized (workers) {
            if (workers.size() == coreSize) {
                // taskQueue.offer(runnable);
                taskQueue.tryOffer(runnable, rejectStrategy);
            } else {
                Worker worker = new Worker(runnable);
                workers.add(worker);
                worker.start();
            }
        }

    }

    @EqualsAndHashCode(callSuper = true)
    @Data
    class Worker extends Thread {
        private Runnable r;

        public Worker(Runnable r) {
            this.r = r;
        }

        @Override
        public void run() {
            while (r != null || (r = taskQueue.poll()) != null) {
                try {
                    r.run();
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    r = null;
                }
            }
            // 退出循环代表线程任务执行结束，需要清理掉改线程
            synchronized (workers) {
                workers.remove(this);
            }
        }
    }
}

class BlockQueue<T> {
    // 1. 任务队列
    private Deque<T> deque = new ArrayDeque<>();
    // 2. 锁
    private Lock lock = new ReentrantLock();
    // 3. 消费者condition
    private Condition consumerCondition = lock.newCondition();
    // 4. 生产者condition
    private Condition producerCondition = lock.newCondition();
    // 5. 容量
    private int capacity;

    public BlockQueue(int capacity) {
        this.capacity = capacity;
    }

    public int size() {
        lock.lock();

        try {
            return deque.size();
        } finally {
            lock.unlock();
        }
    }

    public T poll(int timeout, TimeUnit unit) {
        lock.lock();
        long nanos = unit.toNanos(timeout);
        try {
            while (deque.isEmpty()) {
                try {
                    if (nanos <= 0) {
                        return null;
                    }
                    nanos = consumerCondition.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T poll = deque.poll();
            producerCondition.signal();
            return poll;

        } finally {
            lock.unlock();
        }
    }

    public T poll() {
        lock.lock();
        try {
            while (deque.isEmpty()) {
                try {
                    consumerCondition.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            T poll = deque.poll();
            producerCondition.signal();
            return poll;
        } finally {
            lock.unlock();
        }

    }

    public void offer(T t) {
        lock.lock();
        try {
            while (deque.size() == capacity) {
                try {
                    producerCondition.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            deque.offer(t);
            consumerCondition.signal();
        } finally {
            lock.unlock();
        }

    }

    public void tryOffer(T runnable, RejectStrategy<T> rejectStrategy) {
        lock.lock();
        try {
            if (deque.size() == capacity) {
                rejectStrategy.reject(this, runnable);
            } else {
                deque.offer(runnable);
                consumerCondition.signal();
            }
        } finally {
            lock.unlock();
        }
    }
}
